package com.wulaobo.aggregation;

import com.wulaobo.pojo.WaterSensor;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author wuwenbo
 * @Date 2024/5/17 11:39
 * @ClassName: KeyByDemo
 * @Description: TODO
 */
public class SumDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);


        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1, 1),
                new WaterSensor("sensor_2", 2, 3),
                new WaterSensor("sensor_2", 4, 2),
                new WaterSensor("sensor_4", 3, 4),
                new WaterSensor("sensor_4", 40, 44),
                new WaterSensor("sensor_4", 4, 1),
                new WaterSensor("sensor_4", 4, 5)
        );

        //key by
        KeyedStream<WaterSensor, String> keyedStream = stream.keyBy(e -> e.getId());

        //sum
        SingleOutputStreamOperator<WaterSensor> vc = keyedStream.sum("vc");
        vc.print();

//        keyedStream.min("ts").print();

//        keyedStream.max("vc").print();
//        SingleOutputStreamOperator<WaterSensor> vc = keyedStream.min("vc");
//        vc.print();

        env.execute();

    }

}
